package com.song.sparkstudy;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFlatMapFunction;

import scala.Tuple2;

public class JavaLambaRddTest {
	
	
	
	public static void main(String[] args) {
		SparkConf conf = new SparkConf().setAppName("rddtest").setMaster("local[2]");
		JavaSparkContext context = new JavaSparkContext(conf);
		
		
        JavaRDD<Integer> rdd = context.parallelize(Arrays.asList(1, 2, 3, 4, 5));
        List<String> list = new ArrayList<>();
        list.add("a,b,c,d,e");
        list.add("1,2,3,4,5");
        JavaRDD<String> rddString1 = context.parallelize(list);

        List<String> list2 = new ArrayList<>();
        list2.add("1,2,3,4,5");
        list2.add("aa,bb,cc,dd,ee");
        list2.add("11,22,33,44,55");
        JavaRDD<String> rddString2 = context.parallelize(list2);
		
		rddmap(rdd);
		rddmap2(rdd);
		rddfilter(rdd);
		rddunion(rddString1 , rddString2);
		maptopair(rddString1);
		context.close();
	}

	private static void maptopair(JavaRDD<String> rddString1) {
		

		rddString1.flatMapToPair(x->{
			String[] tmp = x.split(",");
			return Arrays.asList( new Tuple2<String,String>(tmp[0], tmp[1])) ;
			
		}).collect().forEach(t->{
			System.out.println(t._1 +"=------"+ t._2);
		});
		
	}

	private static void rddunion(JavaRDD<String> rddString1, JavaRDD<String> rddString2) {
		JavaRDD<String> unionrdd = rddString1.union(rddString2);
		
		for(String str :unionrdd.collect())
		{
			System.out.println("  union str is "+ str);
		}
		
	}

	private static void rddfilter(JavaRDD<Integer> rdd) {
		
		 rdd.filter(x-> (x %2==0)).collect().forEach(System.out::println);
	}

	private static void rddmap2(JavaRDD<Integer> rdd) {
		

		rdd.map(x->new Tuple2<Integer, Integer>(x, x*3)).collect().forEach(t->{
			System.out.println(t._1+"---->"+t._2);
		});
		
	}

	/**
	 *  对原来的数据进行一定的操作
	 * @param rdd
	 */
	private static void rddmap(JavaRDD<Integer> rdd) {
		

		rdd.map(x -> x*2).collect().forEach(s->{System.out.println(s);});
		
	}
	
	
	
	
}
